#include <assert.h>
#include <fcntl.h>
#include <string.h>
#include <stdio.h>
#include <stdlib.h>
#include <algorithm>
extern "C" {
  #include "ext/urlparser/urlparser.h"
}

#include <iostream>
#include <map>
#include <sstream>
#include <string>
#include <vector>

#include <v8.h>
#include <v8-debug.h>

#include "jsmr.h"

#define JSMRLOG(level, msg)  Log(level, __FILE__, __LINE__, msg);


//TODO(nathan, crutcher): Shouldn't these be in the 'internal' namespace?
// Extracts a C string from a V8 Utf8Value.
static const char* ToCString(const v8::String::Utf8Value& value) {
  return *value ? *value : "<string conversion failed>";
}

static std::string ToString(const v8::String::Utf8Value& value) {
  return std::string(ToCString(value));
}

// TODO(crutcher): treat encoding (utf8, etc) intelligently.
static const v8::Handle<v8::String> ToV8String(const std::string &s) {
  return v8::String::New(s.c_str(), s.length());
}

namespace JSMR {

CharBufferStringResource::CharBufferStringResource(char *data, size_t length)
      : data_(data), length_(length) {}

CharBufferStringResource::~CharBufferStringResource() { delete[] data_; }

const char* CharBufferStringResource::data() const { return data_; }

size_t CharBufferStringResource::length() const { return length_; }


/**
 * An array which has the human readable names of the log levels.  There's a
 * 1-to-1 correspondence with the enum.
 */
//TODO(nathan, crutcher): We *could* have a macro which whill enforce the
//equivalence of the log level name to the output.
static const std::string g_LogLevelNames[] = {
  std::string("FATAL"),
  std::string("ERROR"),
  std::string("WARN"),
  std::string("INFO"),
};

/**
 * Looks up a log level and returns the string containing that name.  Any level
 * less than 0 is FATAL and anything greater than LAST_LOG_LEVEL just becomes
 * LAST_LOG_LEVEL (this has the effect of becoming INFO).
 * @param level a log level to look up
 * @return The stringified version of hte log-level 
 */
const std::string &LogLevelName(LogLevel level) {
  if (level < 0) level = FATAL;
  if (level > LAST_LOG_LEVEL) level = LAST_LOG_LEVEL;
  return g_LogLevelNames[level];
}

void OperationEnvironment::Log(
    LogLevel level,
    const std::string &source,
    unsigned int line,
    const std::string &msg) {
  LogImpl(level, source, line, msg);
  if(level <= FATAL) {
    OnFatal();
    abort();
  }
}

void OperationEnvironment::LogImpl(
    LogLevel level,
    const std::string &source,
    unsigned int line,
    const std::string &msg) {
  std::cerr << "[" << LogLevelName(level) << " "
            << source << ":" << line << "] "
            << msg.c_str() << std::endl;
}

v8::Handle<v8::String> DefaultOperationEnvironment::ReadFile(
    const std::vector<std::string> &search_paths,
    const std::string &path) {
  FILE* file = fopen(path.c_str(), "rb");
  if (file == NULL) {
    // TODO(crutcher): use search_paths
    JSMRLOG(FATAL, "Script file not found: " + path);
  }

  fseek(file, 0, SEEK_END);
  int size = ftell(file);
  rewind(file);

  char* chars = new char[size + 1];
  chars[size] = '\0';
  for (int i = 0; i < size;) {
    int read = fread(&chars[i], 1, size - i, file);
    i += read;
  }
  fclose(file);

  v8::Handle<v8::String> result = v8::String::NewExternal(
    new CharBufferStringResource(chars, size));

  return result;
}

/**
 * Constructor that does nothing
 */
// TODO(nathan) Should this be removed entirely?
ProtocolHandler::ProtocolHandler () { }

/**
 * Abstract class used to assocaite an appropritate 'ReadFile' definition with
 * the correct protocol.  Used to associate a protocol name (e.g.
 * 'http', 'file', 'hdfs', e.g.) with a file reader.
 * @param protocol_name The name of the protocol, e.g. 'hdfs'
 */
ProtocolHandler::ProtocolHandler(std::string protocol_name) :
    protocol_name_(protocol_name) { }

/**
 * Returns the protocol name associated with this protocol.
 * @return The string that's associated with this protocol.
 */
std::string ProtocolHandler::GetProtocolName() {
  return protocol_name_;
}

/**
 * This is a sub-class of the abstract class ProtocolHandler, specifically for
 * reading local files.  It is associated with the protocol name 'file'
 */
LocalFileHandler::LocalFileHandler() : ProtocolHandler("file"){ }

/**
 * Read a file from the local file system.
 * @param long_path The full path to the file on the local file system
 * @return contents of the file read.
 */
v8::Handle<v8::String> LocalFileHandler::ReadFile
    (const std::string &long_path) {
  ::URL storage;
  ::parseURLField(long_path.c_str(), &storage, URL_PATH);

  std::string path(readURLField(long_path.c_str(), storage.path));
  //long_path.substr((GetProtocolName() + "://").length());

  FILE* file = fopen(path.c_str(), "rb");
  if (file == NULL) {
    // TODO(nathan):  Need to figure out how to push the logger into here.
    std::cerr << "Script file not found: " << path << std:: endl;
    abort();
  }

  fseek(file, 0, SEEK_END);
  int size = ftell(file);
  rewind(file);

  char* chars = new char[size + 1];
  chars[size] = '\0';
  for (int i = 0; i < size;) {
    int read = fread(&chars[i], 1, size - i, file);
    i += read;
  }
  fclose(file);

  v8::Handle<v8::String> result = v8::String::NewExternal(
    new CharBufferStringResource(chars, size));

  return result;
}

/**
 * Constructor for JSContainer
 * @param op_environment a pointer to the OperationEnvironment
 * @param config_json A string being passed in which contains JSON defining the
 * run-time configuration.
 */
JSContainer::JSContainer(OperationEnvironment *op_environment,
                         const std::string &config_json)
    : op_environment_(op_environment) {
  v8::HandleScope handle_scope;

  v8::Handle<v8::String> JSMR_string = v8::String::New("JSMR");
  v8::Handle<v8::String> config_string = v8::String::New("config");

  v8::Handle<v8::External> that = v8::External::New(this);

  v8::Handle<v8::ObjectTemplate> global_templ = v8::ObjectTemplate::New();

  v8::Handle<v8::ObjectTemplate> jsmr_templ = v8::ObjectTemplate::New();
  global_templ->Set(JSMR_string, jsmr_templ);

  jsmr_templ->Set(v8::String::New("require"),
      v8::FunctionTemplate::New(&JSContainer::RequireCallback, that));

  v8::Handle<v8::FunctionTemplate> log_templ =
      v8::FunctionTemplate::New(&JSContainer::LogCallback, that);
  jsmr_templ->Set(v8::String::New("log"), log_templ);

  // Install LogLevels on the 'log' object.
  for (int i = 0; i <= LAST_LOG_LEVEL; ++i) {
    const LogLevel level = static_cast<LogLevel>(i);
    log_templ->Set(ToV8String(LogLevelName(level)), v8::Integer::New(level));
  }

  v8::Handle<v8::ObjectTemplate> config_templ = v8::ObjectTemplate::New();
  jsmr_templ->Set(config_string, config_templ);

  v8::Handle<v8::Context> context = v8::Context::New(NULL, global_templ);
  context_ = v8::Persistent<v8::Context>::New(context);

  v8::Context::Scope context_scope(context_);

  // TODO(crutcher): refactor this trash.
  v8::TryCatch try_catch;
  v8::Handle<v8::Script> script = v8::Script::Compile(
      v8::String::New(
      "(function (exec_state, depth) {"
         "var f = exec_state.frame(depth);"
         "return [f.func().script().name(), f.sourceLine() + 1];"
       "})"));
  if (try_catch.HasCaught()) {
    ReportException(FATAL, &try_catch);
  }
  v8::Handle<v8::Value> hook_result = script->Run();
  if (try_catch.HasCaught()) {
    ReportException(FATAL, &try_catch);
  }
  source_array_func_ = v8::Persistent<v8::Function>::New(
      v8::Handle<v8::Function>::Cast(hook_result));

  jsmr_obj_ = v8::Persistent<v8::Object>::New(
      v8::Handle<v8::Object>::Cast(
          context_->Global()->Get(JSMR_string)));

  config_obj_ = v8::Persistent<v8::Object>::New(
      v8::Handle<v8::Object>::Cast(
          ParseJSON(ToV8String(config_json))));
  if (try_catch.HasCaught()) {
    ReportException(FATAL, &try_catch);
  }

  jsmr_obj_->Set(config_string, config_obj_);
}

/**
 * Destructor for a JSContainer.  
 */
JSContainer::~JSContainer() {
  loaded_scripts_.clear();
  source_array_func_.Dispose();
  context_.Dispose();
}

/**
 * Boot the container by running all scripts listed in JSMR.config.scripts
 */
void JSContainer::Boot() {
  v8::HandleScope handle_scope;
  v8::Context::Scope context_scope(context_);
  v8::Handle<v8::Array> scripts = v8::Handle<v8::Array>::Cast(
      config_obj_->Get(v8::String::New("scripts")));
  for (unsigned int i = 0; i < scripts->Length(); ++i) {
    v8::String::Utf8Value script(scripts->Get(i));
    RunScript(ToString(script));
  }
}

/**
 * Private method for reporting exceptions.
 * @param level The log level to report at (e.g. 'WARN')
 * @param try_catch The 'TryCatch' which contains the exception information.
 */
void JSContainer::ReportException(LogLevel level, v8::TryCatch *try_catch) {
  v8::HandleScope handle_scope;
  v8::String::Utf8Value exception(try_catch->Exception());
  v8::Handle<v8::Message> message = try_catch->Message();

  if (message.IsEmpty()) {
    // V8 didn't provide any extra information about this error;
    // just print the exception.
    JSMRLOG(level, ToCString(exception));
  } else {
    std::ostringstream output;
    v8::String::Utf8Value filename(message->GetScriptResourceName());

    int line_num = message->GetLineNumber();
    int column_start = message->GetStartColumn();
    int column_end = message->GetEndColumn();

    output << ToCString(exception) << std::endl;

    std::string prefix = "   ";

    // Print line of source code.
    v8::String::Utf8Value source_line(message->GetSourceLine());
    output << prefix << ToCString(source_line) << std::endl;

    // Print wavy underline (GetUnderline is deprecated).
    for (unsigned int i = 0; i < column_start + prefix.length(); i++) {
      output << " ";
    }

    for (int i = column_start; i < column_end; i++) {
      output << "^";
    }

    output << std::endl;

    op_environment_->Log(
        FATAL, std::string(ToCString(filename)),line_num, output.str());
  }
}

/** 
 * Pass through to the logger from the operation environment.
 * @param level the log level to log at (e.g. 'WARN')
 * @param source the name of the file that 'Log' was called from.
 * @param line the line number where 'Log' was called.
 * @param msg the message to report.
 */
void JSContainer::Log(
    LogLevel level,
    const std::string &source,
    unsigned int line,
    const std::string &msg) {
  op_environment_->Log(level, source, line, msg);
}

/**
 * Private method for parsing a string containing JSON and returning a
 * JavaScript Value which represents it.
 * @param str the string containing the JSON
 * @return the evaluated JSON as a JavaScript value
 */ 
v8::Handle<v8::Value> JSContainer::ParseJSON(v8::Handle<v8::String> str) {
  v8::HandleScope handle_scope;

  // TODO(crutcher): This sorta dies if JSON.parse isn't defined in the
  // environment.  Is that a problem?
  v8::Handle<v8::Object> JSON_obj = v8::Handle<v8::Object>::Cast(
      context_->Global()->Get(v8::String::New("JSON")));
  v8::Handle<v8::Function> parse_func = v8::Handle<v8::Function>::Cast(
      JSON_obj->Get(v8::String::New("parse")));

  const int argc = 1;
  v8::Handle<v8::Value> argv[argc] = { str };
  v8::Handle<v8::Value> result = parse_func->Call(context_->Global(), argc, argv);

  return handle_scope.Close(result);
}

/**
 * Called from within the JavaScript environment when the user calls:
 *    jsmr.log(...)
 * @param args the arguments to the JavaScript function.  These should include
 * the log level, the message, and the stack depth to look up line numbers on.
 * @return v8::Undefined (effectively a void function)
 */
v8::Handle<v8::Value> JSContainer::LogCallback(const v8::Arguments& args) {
  // TODO(crutcher): throw an exception here.
  if (args.Length() < 2) return v8::Undefined();
  v8::HandleScope scope;

  JSContainer *mapper = static_cast<JSContainer *>(
      v8::Handle<v8::External>::Cast(args.Data())->Value());

  LogLevel level = (LogLevel) args[0]->ToInt32()->Value();
  v8::String::Utf8Value message(args[1]);

  v8::Handle<v8::Value> depth;
  if (args.Length() >= 3) {
    depth = args[2]->ToUint32();
  } else {
    depth = v8::Uint32::New(0);
  }

  v8::Handle<v8::Array> source_location = v8::Handle<v8::Array>::Cast(
      v8::Debug::Call(mapper->source_array_func_, depth));
  assert(source_location->Length() == 2);

  v8::String::Utf8Value source_path(source_location->Get(0));
  unsigned int source_line = source_location->Get(1)->ToUint32()->Value();

  mapper->Log(level, ToCString(source_path), source_line, ToCString(message));
  return v8::Undefined();
}

/**
 * This is called by the JavaScript environment when the user calls:
 *    require(...)
 * @param args list of scripts to load (as strings).
 * @return v8::Undefined() (effectively a void function)
 */
v8::Handle<v8::Value> JSContainer::RequireCallback(const v8::Arguments& args) {
  v8::HandleScope scope;

  JSContainer *mapper = static_cast<JSContainer *>(
      v8::Handle<v8::External>::Cast(args.Data())->Value());

  for (int i = 0; i < args.Length(); ++i) {
    v8::String::Utf8Value path(args[i]);
    // TODO(crutcher): type/error checking and reporting.
    mapper->RunScript(*path);
  }
  return v8::Undefined();
}

/**
 * This is called by the JavaScript environment when the user calls:
 *    emit(key, value)
 * calls the Emit method from the current Map Context.
 * @param args should be a key and a value
 * @return v8::Undefined() (essentially a void function)
 */
v8::Handle<v8::Value> JSContainer::EmitCallback(const v8::Arguments& args) {
  v8::HandleScope scope;

  // TODO(nathan,crutcher): Throw an exception upon wrong args
  if(args.Length() != 2) {
    return v8::Undefined();
  }

  MapContext *map_context = static_cast<MapContext *>(
      v8::Handle<v8::External>::Cast(args.Data())->Value());

  v8::String::Utf8Value key(args[0]);
  v8::String::Utf8Value value(args[1]);

  // TODO(crutcher): This breaks in the presence of null characters
  map_context->Emit(std::string(*key), std::string(*value));
  return v8::Undefined();
}

/**
 * Called by 'RequireCallback'.  Evaluates and runs the script that was
 * required.
 * @param path the path passed in via 'require(path)'
 */
void JSContainer::RunScript(const std::string &path) {
  v8::HandleScope handle_scope;
  v8::Context::Scope context_scope(context_);

  if (loaded_scripts_.find(path) != loaded_scripts_.end()) {
    JSMRLOG(INFO, "Not Reloading Script: " + path);
  } else {
    // TODO(crutcher): factor this out and store it in the class.
    v8::Handle<v8::Object> JSMR_obj = v8::Handle<v8::Object>::Cast(
        context_->Global()->Get(v8::String::New("JSMR")));
    v8::Handle<v8::Object> config_obj = v8::Handle<v8::Object>::Cast(
        JSMR_obj->Get(v8::String::New("config")));
    // TODO(crutcher): config.paths is mandatory, but not enforced yet.
    v8::Handle<v8::Array> paths_obj = v8::Handle<v8::Array>::Cast(
        config_obj->Get(v8::String::New("paths")));

    std::vector<std::string> search_paths(paths_obj->Length());

    // TODO(crutcher): ToCString? really?
    for (unsigned int i = 0; i < paths_obj->Length(); ++i) {
      v8::String::Utf8Value path(paths_obj->Get(i));
      search_paths[i] = std::string(ToCString(path));
    }

    loaded_scripts_.insert(path);
    JSMRLOG(INFO, "Loading Script: " + path);
    v8::Handle<v8::String> source =
        op_environment_->ReadFile(search_paths, path);

    v8::TryCatch try_catch;
    v8::Handle<v8::Script> script =
        v8::Script::Compile(source, ToV8String(path));
    if (script.IsEmpty()) {
      ReportException(FATAL, &try_catch);
    } else {
      v8::Handle<v8::Value> result = script->Run();
      if (result.IsEmpty()) {
        ReportException(FATAL, &try_catch);
      }
    }
  }
}

void JSContainer::RunMap(MapContext *map_context) {
  v8::HandleScope handle_scope;
  v8::Context::Scope context_scope(context_);

  // TODO(crutcher): abstract this code out.
  v8::Handle<v8::Value> f_val =
      context_->Global()->Get(v8::String::New("Map"));
  if (f_val->IsUndefined()) {
    JSMRLOG(FATAL,
        "'Map' function is not defined in the script environment.");
  }
  if (!f_val->IsFunction()) {
    JSMRLOG(FATAL,
        "'Map' object is defined, "
        "but is not a function in the script environment.");
  }
  v8::Handle<v8::Function> f = v8::Handle<v8::Function>::Cast(f_val);

  v8::Handle<v8::Object> input_obj = v8::Object::New();
  input_obj->Set(
      v8::String::New("value"), ToV8String(map_context->GetInput()));

  // TODO(nathan,crutcher):  This is not safe if someone reuses the output
  // object

  v8::Handle<v8::External> map_context_obj = v8::External::New(map_context);
  v8::Handle<v8::FunctionTemplate> emit_templ =
      v8::FunctionTemplate::New(&JSContainer::EmitCallback, map_context_obj);
  v8::Handle<v8::Object> output_obj = v8::Object::New();
  output_obj->Set(v8::String::New("emit"), emit_templ->GetFunction());

  v8::TryCatch try_catch;
  const int argc = 2;
  v8::Handle<v8::Value> argv[argc] = { input_obj, output_obj };
  v8::Handle<v8::Value> result = f->Call(context_->Global(), argc, argv);

  // TODO(crutcher): handle/signal per-map failures.
  if (try_catch.HasCaught()) {
    ReportException(FATAL, &try_catch);
  }
}

void JSContainer::RunReduce(ReduceContext *reduce_context) {
  v8::HandleScope handle_scope;
  v8::Context::Scope context_scope(context_);

  // TODO(crutcher): abstract this code out.
  v8::Handle<v8::Value> f_val =
      context_->Global()->Get(v8::String::New("Reduce"));
  if (f_val->IsUndefined()) {
    JSMRLOG(FATAL,
        "'Reduce' function is not defined in the script environment.");
  }
  if (!f_val->IsFunction()) {
    JSMRLOG(FATAL,
        "'Reduce' object is defined, "
        "but is not a function in the script environment.");
  }
  v8::Handle<v8::Function> f = v8::Handle<v8::Function>::Cast(f_val);

  v8::Handle<v8::Object> input_obj = v8::Object::New();
  v8::Handle<v8::Array> input_value_obj = v8::Array::New();

  // TODO(crutcher,nathan): Fix double copy, and handling of null chars.
  int i = 0;
  while(reduce_context->NextValue()) {
    input_value_obj->Set(i++, ToV8String(reduce_context->GetInput()));
  }

  input_obj->Set(
      v8::String::New("key"), ToV8String(reduce_context->GetKey()));
  input_obj->Set(
      v8::String::New("values"),
      input_value_obj);


  // TODO(nathan,crutcher):  This is not safe if someone reuses the output
  // object

  v8::Handle<v8::External> reduce_context_obj = v8::External::New(reduce_context);
  v8::Handle<v8::FunctionTemplate> emit_templ =
      v8::FunctionTemplate::New(&JSContainer::EmitCallback, reduce_context_obj);
  v8::Handle<v8::Object> output_obj = v8::Object::New();
  output_obj->Set(v8::String::New("emit"), emit_templ->GetFunction());

  v8::TryCatch try_catch;
  const int argc = 2;
  v8::Handle<v8::Value> argv[argc] = { input_obj, output_obj };
  v8::Handle<v8::Value> result = f->Call(context_->Global(), argc, argv);

  // TODO(crutcher): handle/signal per-reduce failures.
  if (try_catch.HasCaught()) {
    ReportException(FATAL, &try_catch);
  }
}
}  // namespace JSMR
